package com.example.literj.kt.internal.schedulers

import com.example.literj.kt.core.Scheduler
import com.example.literj.kt.disposables.Disposable
import com.example.literj.kt.internal.disposables.CompositeDisposable
import com.example.literj.kt.internal.disposables.EmptyDisposable
import java.util.concurrent.*
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicReference

class IoScheduler() : Scheduler() {
    constructor(threadFactory: ThreadFactory?) : this() {
        this.threadFactory = WORKER_THREAD_FACTORY
        this.pool = AtomicReference<CachedWorkerPool>(NONE)
        start()
    }

    private var threadFactory: ThreadFactory? = null
    private var pool: AtomicReference<CachedWorkerPool>? = null

    private val WORKER_THREAD_NAME_PREFIX = "RxCachedThreadScheduler"
    private val WORKER_THREAD_FACTORY: RxThreadFactory

    private val EVICTOR_THREAD_NAME_PREFIX = "RxCachedWorkerPoolEvictor"
    private val EVICTOR_THREAD_FACTORY: RxThreadFactory

    private val KEY_KEEP_ALIVE_TIME = "rx3.io-keep-alive-time"
    private val KEEP_ALIVE_TIME_DEFAULT = 60L

    private val KEEP_ALIVE_TIME: Long =
        java.lang.Long.getLong(KEY_KEEP_ALIVE_TIME, KEEP_ALIVE_TIME_DEFAULT) ?: 0L
    private val KEEP_ALIVE_UNIT = TimeUnit.SECONDS

    private val SHUTDOWN_THREAD_WORKER: ThreadWorker =
        ThreadWorker(RxThreadFactory("RxCachedThreadSchedulerShutdown"))

    private val KEY_IO_PRIORITY = "rx3.io-priority"

    private val KEY_SCHEDULED_RELEASE = "rx3.io-scheduled-release"
    private var USE_SCHEDULED_RELEASE = false
    private val NONE: CachedWorkerPool?

    init {
        SHUTDOWN_THREAD_WORKER.dispose()

        val priority = Thread.MIN_PRIORITY.coerceAtLeast(
            Thread.MAX_PRIORITY.coerceAtMost(
                Integer.getInteger(
                    KEY_IO_PRIORITY,
                    Thread.NORM_PRIORITY
                ) ?: 0
            )
        )

        WORKER_THREAD_FACTORY =
            RxThreadFactory(WORKER_THREAD_NAME_PREFIX, priority)

        EVICTOR_THREAD_FACTORY =
            RxThreadFactory(EVICTOR_THREAD_NAME_PREFIX, priority)

        USE_SCHEDULED_RELEASE = java.lang.Boolean.getBoolean(KEY_SCHEDULED_RELEASE)

        NONE = CachedWorkerPool(0, null, WORKER_THREAD_FACTORY)
        NONE.shutdown()
    }

    override fun start() {
        val update = CachedWorkerPool(KEEP_ALIVE_TIME, KEEP_ALIVE_UNIT, threadFactory!!)
        if (!pool!!.compareAndSet(NONE, update)) {
            update.shutdown()
        }
    }

    override fun shutdown() {
        val cur = pool?.getAndSet(NONE)
        if (cur != NONE) {
            cur?.shutdown()
        }
    }

    override fun createWorker() = EventLoopWorker(pool!!.get())

    inner class CachedWorkerPool(
        keepAliveTime: Long,
        unit: TimeUnit?,
        private val threadFactory: ThreadFactory
    ) :
        Runnable {
        private val keepAliveTime: Long
        private val expiringWorkerQueue: ConcurrentLinkedQueue<ThreadWorker>
        private val allWorkers: CompositeDisposable
        private val evictorService: ScheduledExecutorService?
        private val evictorTask: Future<*>?

        init {
            this.keepAliveTime = unit?.toNanos(keepAliveTime) ?: 0L
            this.expiringWorkerQueue = ConcurrentLinkedQueue()
            this.allWorkers = CompositeDisposable()
            var evictor: ScheduledExecutorService? = null
            var task: Future<*>? = null
            if (unit != null) {
                evictor = Executors.newScheduledThreadPool(1, EVICTOR_THREAD_FACTORY)
                task = evictor.scheduleWithFixedDelay(
                    this,
                    this.keepAliveTime,
                    this.keepAliveTime,
                    TimeUnit.NANOSECONDS
                )
            }
            evictorService = evictor
            evictorTask = task
        }

        override fun run() {
            evictExpiredWorkers(expiringWorkerQueue, allWorkers)
        }

        fun get(): ThreadWorker {
            if (allWorkers.isDisposed()) {
                return SHUTDOWN_THREAD_WORKER
            }
            while (!expiringWorkerQueue.isEmpty()) {
                val threadWorker: ThreadWorker? = expiringWorkerQueue.poll()
                if (threadWorker != null) {
                    return threadWorker
                }
            }

            val w = ThreadWorker(threadFactory)
            allWorkers.add(w)
            return w
        }

        fun release(threadWorker: ThreadWorker) {
            threadWorker.expirationTime = now() + keepAliveTime
            expiringWorkerQueue.offer(threadWorker)
        }

        fun shutdown() {
            allWorkers.dispose()
            evictorTask?.cancel(true)
            evictorService?.shutdownNow()
        }

        fun evictExpiredWorkers(
            expiringWorkerQueue: ConcurrentLinkedQueue<ThreadWorker>,
            allWorkers: CompositeDisposable
        ) {
            if (!expiringWorkerQueue.isEmpty()) {
                val currentTimestamp = now()
                for (threadWorker in expiringWorkerQueue) {
                    if (threadWorker.expirationTime <= currentTimestamp) {
                        if (expiringWorkerQueue.remove(threadWorker)) {
                            allWorkers.remove(threadWorker)
                        }
                    } else {
                        break
                    }
                }
            }
        }

        fun now(): Long {
            return System.nanoTime()
        }
    }

    inner class EventLoopWorker(private val pool: CachedWorkerPool) : Worker(),
        Runnable {
        private val tasks: CompositeDisposable = CompositeDisposable()
        private val threadWorker: ThreadWorker = pool.get()
        private val once = AtomicBoolean()

        override fun dispose() {
            if (once.compareAndSet(false, true)) {
                tasks.dispose()
                if (USE_SCHEDULED_RELEASE) {
                    threadWorker.scheduleActual(this, 0, TimeUnit.NANOSECONDS, null)
                } else {
                    pool.release(threadWorker)
                }
            }
        }

        override fun run() {
            pool.release(threadWorker)
        }

        override fun isDisposed(): Boolean {
            return once.get()
        }

        override fun schedule(run: Runnable?, delay: Long, unit: TimeUnit?): Disposable {
            return if (tasks.isDisposed()) {
                EmptyDisposable.INSTANCE
            } else threadWorker.scheduleActual(run, delay, unit, tasks)
        }
    }

    inner class ThreadWorker(threadFactory: ThreadFactory) : NewThreadWorker(threadFactory) {
        var expirationTime = 0L
    }
}